#[macro_use]
extern crate log;

use std::thread;
use std::time::Duration;

use anyhow::{bail, Result};
use smol::net::{TcpListener, TcpStream};
use smol::prelude::{AsyncReadExt, AsyncWriteExt, FutureExt};
use smol::stream::StreamExt;
use smol::Timer;

fn main() {
    std::env::set_var("RUST_LOG", "trace");
    pretty_env_logger::init();

    info!("{:?} => program start", thread::current().id());
    match smol::block_on(run()) {
        Ok(_) => info!("Success"),
        Err(e) => error!("{}", e),
    }
}

async fn run() -> Result<()> {
    let listener = TcpListener::bind("0.0.0.0:9000").await?;

    info!("{:?} => listen on 0.0.0.0:9000", thread::current().id());
    while let Some(stream) = listener.incoming().next().await {
        let stream = stream?;
        smol::spawn(handle(stream)).detach();
    }

    Ok(())
}

async fn handle(mut stream: TcpStream) -> Result<()> {
    let req = read_tcp(stream.clone()).await?;
    info!("{:?} => receive req: {}", thread::current().id(), req);

    let mut rpc_client = TcpStream::connect("127.0.0.1:9100").await?;
    let res = call_service(rpc_client.clone(), req).await?;
    rpc_client.close().await?;

    let res = format!("{:010}{}", res.len(), res);
    stream.write_all(res.as_bytes()).await?;
    stream.flush().await?;
    stream.close().await?;
    info!("{:?} => send res: {}", thread::current().id(), res);
    Ok(())
}

async fn call_service(mut rpc_stream: TcpStream, req: impl ToString) -> Result<String> {
    let req = req.to_string();
    debug!("{:?} => rpc start", thread::current().id());

    //请求
    let req = format!("{:010}{}", req.len(), req);
    rpc_stream.write_all(req.as_bytes()).await?;
    rpc_stream.flush().await?;
    debug!("{:?} => rpc done", thread::current().id());

    //响应
    let res = read_tcp(rpc_stream).await?;
    Ok(res)
}

async fn read_tcp(mut stream: TcpStream) -> Result<String> {
    let timeout = |t| async move {
        Timer::after(Duration::from_secs(t)).await;
        Ok(())
    };

    //读10位长度
    let mut len = [0_u8; 10];
    stream.read_exact(&mut len).or(timeout(30)).await?;
    let length: usize = String::from_utf8_lossy(&len).parse()?;
    if length < 1 {
        warn!(
            "{:?} => receive length less then 1: {}",
            thread::current().id(),
            length
        );
        bail!("length is 0")
    }

    //读指定长度的报文
    let mut msg = vec![0_u8; length];
    stream.read_exact(&mut msg).or(timeout(30)).await?;
    let msg = String::from_utf8(msg)?;

    Ok(msg)
}
